package net.bwie.realtime.dws.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.dws.log.bean.AreaControl;
import net.bwie.realtime.dws.log.function.PageViewReportReduceFunction;
import net.bwie.realtime.dws.log.function.PageViewReportWindowFunction;
import net.bwie.realtime.dws.log.function.TrafficMapFunction;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import utils.DorisUtil;
import utils.KafkaUtil;

public class VehicleDistribution {

    public static void main(String[] args) throws Exception{

        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        //2.读取数据源
        DataStream<String> carStream = KafkaUtil.consumerKafka(env, "dwd-traffic-monitor-log");

        //3.数据处理
        DataStream<String> resultStream = handle(carStream);
        resultStream.print("result");
        //4.数据输出
        DorisUtil.saveToDoris(resultStream,
                "vehicle_realtime_report", "dws_vehicle_car_distribution_report");

        //5.执行
        env.execute("VehicleDistribution");
    }

    private static DataStream<String> handle(DataStream<String> stream) {
        //按照areaId进行分组，用于计算车辆去重
        KeyedStream<String, String> areaIdStream = stream.keyBy(
                json -> JSON.parseObject(json).getString("areaId") + "," +
                        JSON.parseObject(json).getString("car")
        );

        // 将流中每条数据进行封装bean对象
        SingleOutputStreamOperator<AreaControl> beanStream = areaIdStream.map(new TrafficMapFunction());

        //设置分组
        KeyedStream<AreaControl, String> keyedStream = beanStream.keyBy(bean -> bean.getAreaId());

        //窗口 10分钟
        WindowedStream<AreaControl, String, TimeWindow> windowStream = keyedStream.window(
                TumblingProcessingTimeWindows.of(Time.minutes(1))
        );

        //聚合计算
        DataStream<String> reportStream = windowStream.reduce(
                new PageViewReportReduceFunction(),new PageViewReportWindowFunction()
        );

        return reportStream;
    }
}
